/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.asp.bigdata.flink.kafka2hdfs.task;

import com.asp.bigdata.flink.kafka2hdfs.gz.GzipBulkStringWriterFactory;
import com.asp.bigdata.flink.kafka2hdfs.pojo.LgTyrzAnalysis;
import com.asp.bigdata.flink.kafka2hdfs.utils.CustomBucketAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.sql.Timestamp;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;

@Deprecated
public class StreamingJobTest {

	// Kafka broker list
	private static String BROKER_LIST = "10.169.129.192:9092";
	// Kafka topic
	private static String TOPIC = "test1";
	// Kafka group ID
	private static String GROUP_ID = "lg_rhAnalysis_consumer_flink_gz_test2";

	private  static String PATH = "hdfs:///warehouse/tablespace/managed/hive/lg_tyrz_db03.db/test1";

	public static void main(String[] args) throws Exception {

		BROKER_LIST = args[0];
		TOPIC = args[1];
		GROUP_ID = args[2];

		// set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.enableCheckpointing(100000);
		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(200, 30000));

		CheckpointConfig checkpointConfig = env.getCheckpointConfig();
		checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
		// 任务取消时清除外部保存的检查点(非暴力停止如yarn kill)
		checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
		checkpointConfig.setTolerableCheckpointFailureNumber(10);

		Properties properties = new Properties();
		properties.setProperty("bootstrap.servers", BROKER_LIST);
		properties.setProperty("group.id", GROUP_ID);
//		properties.setProperty("security.protocol", "SASL_PLAINTEXT");
//		properties.setProperty("sasl.mechanism", "PLAIN");
		// properties.setProperty("sasl.kerberos.service.name", "kafka");

		FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>(TOPIC, new SimpleStringSchema(), properties);
		consumer.setStartFromGroupOffsets();

		DataStreamSource<String> streamSource = env.addSource(consumer, "Kafka");

		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

		// 配置watermark，指定事件时间列
		SingleOutputStreamOperator<LgTyrzAnalysis> operatorWithWM = streamSource.flatMap(
				(FlatMapFunction<String, LgTyrzAnalysis>) (value, out) -> {
					String[] fields = value.split("\\|");
					out.collect(new LgTyrzAnalysis(fields[0], fields[1], fields[2], fields[3], fields[4], fields[5]
							, fields[6], fields[7], fields[8], fields[9]
							, fields[10], fields[11], fields[12], fields[13]));
				})
				.returns(TypeInformation.of(LgTyrzAnalysis.class))
				.assignTimestampsAndWatermarks(WatermarkStrategy
						.<LgTyrzAnalysis>forBoundedOutOfOrderness(Duration.ofMinutes(5))
						.withTimestampAssigner((event, timestamp) -> {
							try {
								return Timestamp.valueOf(format.format(format.parse(event.getRequestDate()))).getTime();
							} catch (ParseException e) {
								e.printStackTrace();
							}
							return Timestamp.valueOf("1970-01-01 00:00:00").getTime();
						})  // 指定事件时间列
						.withIdleness(Duration.ofMinutes(1))  // 1分钟内仍未收到数据，标记为空闲数据源
				);

		// 转换成String
		SingleOutputStreamOperator<String> operatorString = operatorWithWM.map(
				(MapFunction<LgTyrzAnalysis, String>) value -> value.toString()
		);

		// 添加gz后缀
		OutputFileConfig config = OutputFileConfig.builder().withPartSuffix(".gz").build();

		final StreamingFileSink<String> hdfsSink = StreamingFileSink
				.forBulkFormat(
						new Path(PATH),
						new GzipBulkStringWriterFactory<String>())
				.withBucketAssigner(new CustomBucketAssigner())
				.withRollingPolicy(OnCheckpointRollingPolicy.build())
				.withOutputFileConfig(config)
				.build();

		operatorString.addSink(hdfsSink).name("hdfs");

		// execute program
		env.execute("StreamingJobTest");
	}
}
